package org.fuys.etp.core;

import javafx.util.Pair;
import lombok.extern.slf4j.Slf4j;
import org.fuys.etp.core.checker.EtpUpdateCheckerChain;
import org.fuys.etp.core.metrics.EtpMetricsHandlerChain;
import org.fuys.etp.core.model.event.EtpReportEvent;
import org.fuys.etp.core.model.req.TpStrategyChangeReq;
import org.fuys.etp.core.model.res.*;
import org.fuys.etp.core.model.type.EtpTypeEnum;
import org.fuys.etp.core.model.vo.TpBaseConfigVO;
import org.fuys.etp.core.model.vo.TpConfigVO;
import org.fuys.etp.core.model.vo.TpRegulationStrategyVO;
import org.fuys.etp.core.pool.EtpThreadPoolExecutor;
import org.fuys.etp.core.pool.factory.NamedThreadPoolFactory;
import org.fuys.etp.core.pool.queue.ResizeableLinkedBlockingQueue;
import org.fuys.etp.core.reject.EtpRejectHandler;
import org.fuys.etp.repo.EtpRepoService;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;

/**
 * @projectName: fuys-etp-final
 * @package: org.fuys.etp.core
 * @className: EtpManager
 * @author: WangJing
 * @description: TODO
 * @date: 2025/2/18 16:00
 * @version: 1.0
 */
@Component
@Slf4j
public class EtpManager {
    //k->线程池类型,线程池名称 v->线程池实例
    Map<Pair<String,String>, ThreadPoolExecutor> executorMap;

    //k->线程池类型,线程池名称 v->线程池锁
    Map<Pair<String,String>, ReentrantReadWriteLock> lockMap;

    //k->线程池类型,线程池名称 v->线程池策略
    Map<Pair<String,String>, TpRegulationStrategyVO> strategyMap;

    //k->策略ID v->线程池策略
    Map<Integer,TpRegulationStrategyVO> strategyIdMap;

    //k->线程池类型,配置版本号 v->线程池配置
    Map<Pair<String,Integer>, TpConfigVO> configMap;

    Map<String, EtpUpdateCheckerChain> checkerChainMap;

    Map<String, EtpMetricsHandlerChain> metricsChainMap;

    EtpPersistentConfig globalConfig;

    ApplicationEventPublisher eventPublisher;

    EtpRepoService etpRepoService;

    ListableBeanFactory beanFactory;

    ThreadPoolExecutor persistentExecutor;


    public EtpManager(EtpPersistentConfig globalConfig,
                      EtpRepoService repoService,
                      ListableBeanFactory beanFactory,
                      ApplicationEventPublisher eventPublisher){
        this.globalConfig = globalConfig;
        this.etpRepoService = repoService;
        this.eventPublisher = eventPublisher;
        this.beanFactory = beanFactory;
        this.metricsChainMap = beanFactory.getBeansOfType(EtpMetricsHandlerChain.class);
        if(ObjectUtils.isEmpty(metricsChainMap)){
            log.warn("Fuys-Etp: 没有检测到任何指标处理器 这可能会导致一些问题");
        }
        this.checkerChainMap = beanFactory.getBeansOfType(EtpUpdateCheckerChain.class);
        if(ObjectUtils.isEmpty(checkerChainMap)){
            log.warn("Fuys-Etp: 没有检测到任何更新检查器 这可能会导致一些问题");
        }
        executorMap = new ConcurrentHashMap<>();
        lockMap = new ConcurrentHashMap<>();
        strategyIdMap = new ConcurrentHashMap<>();
        strategyMap = new ConcurrentHashMap<>();
        configMap = new ConcurrentHashMap<>();
        if(Boolean.parseBoolean(globalConfig.getHistory())){
            persistentExecutor = createPersistentExecutor(globalConfig);
        }
    }

    public TpCreateResult createExecutor(String poolName,TpBaseConfigVO baseConfig){
        try {
            executorMap.compute(new Pair<>(EtpTypeEnum.COMMON_THREAD_POOL_EXECUTOR.type(), poolName),(k,v)->{
                if(v == null){
                    lockMap.put(k,new ReentrantReadWriteLock());
                    //只是一个普通的线程池 则装填一下就返回即可
                    return new ThreadPoolExecutor(baseConfig.getPoolCoreSize(),baseConfig.getPoolMaxSize(),
                                baseConfig.getPoolKeepAliveTime().longValue(), TimeUnit.valueOf(baseConfig.getPoolAliveTimeUnit().toString()),
                                new LinkedBlockingQueue<>(baseConfig.getWorkQueueSize()),new NamedThreadPoolFactory(poolName),
                                beanFactory.getBean(baseConfig.getRejectHandleBean(),EtpRejectHandler.class));
                }
                log.warn("Fuys Etp: 已存在相同的线程池类型 创建请求被忽略 类型为:{} 名称为:{}",EtpTypeEnum.COMMON_THREAD_POOL_EXECUTOR.type()
                        ,poolName);
                throw new RuntimeException("Fuys Etp: 已存在相同的线程池类型 类型为"+EtpTypeEnum.COMMON_THREAD_POOL_EXECUTOR.type()
                        +" 名称为"+poolName);
            });
            return TpCreateResult.builder().success(true).executor(executorMap.get(new Pair<>(poolName,
                    EtpTypeEnum.COMMON_THREAD_POOL_EXECUTOR.type()))).build();
        }  catch (Exception e) {
            e.printStackTrace();
            lockMap.remove(new Pair<>(poolName, EtpTypeEnum.COMMON_THREAD_POOL_EXECUTOR.type()));
            strategyMap.remove(new Pair<>(poolName, EtpTypeEnum.COMMON_THREAD_POOL_EXECUTOR.type()));
            return TpCreateResult.builder().success(false).extInfo(e.getMessage()).build();
        }
    }

    public TpCreateResult createExecutor(String poolType,String poolName,String strategyId){
        TpRegulationStrategyVO strategyVO = etpRepoService.selectStrategyById(strategyId);
        return this.createExecutor(poolType,poolName,strategyVO);
    }

    public TpCreateResult createExecutor(String poolType,String poolName,TpRegulationStrategyVO strategy){
        //无需启用持久化设置 提供详细策略即可创建
        try {
            final List<Integer> loadConfigIds = strategy.getLoadChain();
            List<TpConfigVO> configVOS = etpRepoService.selectConfigByIds(poolType,loadConfigIds);
            if(ObjectUtils.isEmpty(configVOS)){
                return TpCreateResult.builder().success(false).extInfo("Fuys Etp: 获取到的数据库中配置为空").build();
            }
            sortedByLoadLevel(poolType,configVOS,strategy);
            Pair<TpConfigVO,TpConfigVO> chooseConfig = switchAndSaveConfig(poolType,strategy);
            executorMap.compute(new Pair<>(poolType, poolName),(k,v)->{
                if(v==null){
                    lockMap.put(k,new ReentrantReadWriteLock());
                    strategyMap.put(k,strategy);
                    return createTypeTp(poolName,poolType,strategy.getAlertType(),chooseConfig.getKey(),
                            chooseConfig.getValue(),strategy.getMetricsChain());
                }
                log.warn("Fuys Etp: 已存在相同的线程池类型 创建请求被忽略 类型为:{} 名称为:{}",poolType,poolName);
                throw new RuntimeException("Fuys Etp: 已存在相同的线程池类型 类型为"+poolType+" 名称为"+poolName);
            });
            return TpCreateResult.builder().success(true).executor(executorMap.get(new Pair<>(poolType, poolName))).build();
        } catch (Exception e) {
            e.printStackTrace();
            lockMap.remove(new Pair<>(poolType, poolName));
            strategyMap.remove(new Pair<>(poolType, poolName));
            return TpCreateResult.builder().success(false).extInfo(e.getMessage()).build();
        }
    }

    private ThreadPoolExecutor createTypeTp(String poolName, String poolType, String alertType
            ,TpConfigVO chooseConfig,TpConfigVO limitConfig
            ,String metricsChain) {
        final TpBaseConfigVO baseConfig = chooseConfig.getBaseConfig();
        if(EtpTypeEnum.STAGE_SWITCH_THREAD_POOL_EXECUTOR.type().equals(poolType)
        || EtpTypeEnum.ADAPTIVE_THREAD_POOLS_EXECUTOR.type().equals(poolType)){
            final EtpThreadPoolExecutor etpThreadPoolExecutor = new EtpThreadPoolExecutor(baseConfig.getPoolCoreSize(), baseConfig.getPoolMaxSize(),
                    baseConfig.getPoolKeepAliveTime().longValue(), TimeUnit.valueOf(baseConfig.getPoolAliveTimeUnit().toString()),
                    new LinkedBlockingQueue<>(baseConfig.getWorkQueueSize()), new NamedThreadPoolFactory(poolName),
                    beanFactory.getBean(baseConfig.getRejectHandleBean(),EtpRejectHandler.class));
            etpThreadPoolExecutor.initExecutor(poolName,poolType,alertType,limitConfig,chooseConfig,
                    eventPublisher,metricsChainMap.get(metricsChain));
            return etpThreadPoolExecutor;
        }
        throw new IllegalArgumentException("Fuys etp:不合法参数 没有类型为:"+poolType+"的线程池实现");
    }

    private void sortedByLoadLevel(String poolType,List<TpConfigVO> configVOS, TpRegulationStrategyVO strategy) {
        configVOS.sort(new Comparator<TpConfigVO>() {
            @Override
            public int compare(TpConfigVO o1, TpConfigVO o2) {
                return o1.getLoadLevel().compareTo(o2.getLoadLevel());
            }
        });
        if(EtpTypeEnum.STAGE_SWITCH_THREAD_POOL_EXECUTOR.type().equals(poolType)){
            strategy.setSortedConfigIds(configVOS.stream().map(item->{
                return item.getFixedConfig().getFixedVersion();
            }).collect(Collectors.toList()));
        }else if (EtpTypeEnum.ADAPTIVE_THREAD_POOLS_EXECUTOR.type().equals(poolType)){
            strategy.setSortedConfigIds(configVOS.stream().map(item->{
                return item.getDynamicConfig().getDynamicVersion();
            }).collect(Collectors.toList()));
        }
    }

    private Pair<TpConfigVO,TpConfigVO> switchAndSaveConfig(String poolType
            ,TpRegulationStrategyVO strategy) {
        final Object startLoadLevel = strategy.getStartLoadLevel();
        final List<TpConfigVO> configVOS = strategy.getLoadConfigChain();
        Pair<TpConfigVO,TpConfigVO> result = null;
        final boolean fixed = EtpTypeEnum.STAGE_SWITCH_THREAD_POOL_EXECUTOR.type().equals(poolType);
        final boolean dynamic = EtpTypeEnum.ADAPTIVE_THREAD_POOLS_EXECUTOR.type().equals(poolType);
        for (TpConfigVO configVO : configVOS) {
            if(startLoadLevel!=null){
                if(configVO.getLoadLevel().equals(Integer.valueOf(startLoadLevel.toString()))){
                    final int i = configVOS.indexOf(configVO);
                    if(i == configVOS.size()-1){
                        result = new Pair<>(configVO,configVO);
                    }else{
                        result = new Pair<>(configVO,configVOS.get(i+1));
                    }
                }
            }
            if(fixed){
                configMap.put(new Pair<>(EtpTypeEnum.STAGE_SWITCH_THREAD_POOL_EXECUTOR.type(),
                        configVO.getFixedConfig().getFixedVersion()),configVO);
            }else if(dynamic){
                configMap.put(new Pair<>(EtpTypeEnum.ADAPTIVE_THREAD_POOLS_EXECUTOR.type(),
                        configVO.getDynamicConfig().getDynamicVersion()),configVO);
            }else {
                configMap.put(new Pair<>(EtpTypeEnum.COMMON_THREAD_POOL_EXECUTOR.type(),
                        configVO.getBaseConfig().getVersion()),configVO);
            }
        }
        if(result!=null){
            return result;
        }

        final Pair<TpConfigVO, TpConfigVO> tpConfigVOTpConfigVOPair = new Pair<>(configVOS.get((configVOS.size() - 1) / 2), configVOS.get(Math.min((configVOS.size() - 1) / 2 + 1
                , configVOS.size() - 1)));
        strategy.setStartLoadLevel(tpConfigVOTpConfigVOPair.getKey().getLoadLevel());
        return tpConfigVOTpConfigVOPair;
    }

    public TpConfigUpdateRes updateByCheck(String poolName, String poolType, EtpReportEvent event) {
        log.info("Fuys etp:线程池自检配置中:名称:{},类型:{},自检信息:{}",poolName,poolType,event);
        final TpStatusResult report = event.getReport();
        final TpRegulationStrategyVO strategy = strategyMap.get(new Pair<>(poolType, poolName));
        if(strategy == null || strategy.getCheckerChain() == null){
            //为空 没有策略 无需检查
            return TpConfigUpdateRes.builder().success(false).extInfo("没有对应策略").build();
        }
        final EtpUpdateCheckerChain checkerChain = checkerChainMap.get(strategy.getCheckerChain());
        if(checkerChain == null){
            throw new IllegalArgumentException("Fuys etp:没有类型为:"+strategy.getCheckerChain()+"的检查链路 请检查配置");
        }
        final TpConfigCheckResult tpConfigCheckResult = checkerChain.checkHandle(report, strategy);
        if(tpConfigCheckResult.isNeedChange()){
            //如果需要切换 进行具体的更改
            if(EtpTypeEnum.ADAPTIVE_THREAD_POOLS_EXECUTOR.type().equals(poolType)){
                return this.updateConfig(poolName,poolType,tpConfigCheckResult.getSuggestConfig(),tpConfigCheckResult.getLimitConfig(),
                        null,false);
            }else if(tpConfigCheckResult.getSuggestConfig()!=null){
                return this.updateConfig(poolName,poolType,tpConfigCheckResult.getSuggestConfig(),tpConfigCheckResult.getSuggestConfig(),
                        null,false);
            }
        }
        return TpConfigUpdateRes.builder().success(false).extInfo("无需更改配置").build();
    }

    public TpConfigUpdateRes updateByManual(String poolName,String poolType,TpConfigVO tpConfigVO,Long userId){
        log.info("Fuys etp:线程池人工接管配置中:名称:{},类型:{},配置信息:{}",poolName,poolType,tpConfigVO);
        this.updateConfig(poolName,poolType,tpConfigVO,tpConfigVO,userId,true);
        return TpConfigUpdateRes.normal();
    }

    public TpConfigUpdateRes releaseByManual(String poolName,String poolType,
                                             Integer setConfigId,Integer limitConfigId) throws IllegalAccessException {
        try {
            log.info("Fuys etp:线程池人工取消接管中:名称:{},类型:{},设置的配置ID:{},用于限制的配置ID:{}"
                    ,poolName,poolType,setConfigId,limitConfigId);
            final ThreadPoolExecutor executor = this.executorMap.get(new Pair<>(poolType, poolName));
            final TpConfigVO tpConfigVO = this.configMap.get(new Pair<>(poolType, setConfigId));
            final TpConfigVO limitConfig = this.configMap.get(new Pair<>(poolType, limitConfigId));
            if(tpConfigVO == null || limitConfig == null){
                return TpConfigUpdateRes.builder().success(false).extInfo("取消接管失败 因为指定配置无法获取")
                        .build();
            }
            if(executor instanceof EtpThreadPoolExecutor){
                EtpThreadPoolExecutor etp = (EtpThreadPoolExecutor) executor;
                etp.updateConfig(tpConfigVO,limitConfig);
                etp.relieveTakeover();
            }
            return TpConfigUpdateRes.normal();
        } catch (Exception e) {
            e.printStackTrace();
            return TpConfigUpdateRes.builder().success(false).extInfo(e.getMessage()).build();
        }
    }

    private TpConfigUpdateRes updateConfig(String poolName, String poolType, TpConfigVO suggestConfig,
                                           TpConfigVO limitConfig, Long userId,boolean takeover) {
        try {
            final ThreadPoolExecutor threadPoolExecutor = executorMap.get(new Pair<>(poolType, poolName));
            if(threadPoolExecutor == null){
                return TpConfigUpdateRes.builder().extInfo("不存在此线程池").success(false).build();
            }
            final TpBaseConfigVO baseConfig = suggestConfig.getBaseConfig();
            final TpBaseConfigVO limitBaseConfig = limitConfig.getBaseConfig();
            baseConfig.setRejectHandle(beanFactory.getBean(baseConfig.getRejectHandleBean(),EtpRejectHandler.class));
            limitBaseConfig.setRejectHandle(beanFactory.getBean(baseConfig.getRejectHandleBean(),EtpRejectHandler.class));
            if(threadPoolExecutor instanceof EtpThreadPoolExecutor){
                final EtpThreadPoolExecutor etp = (EtpThreadPoolExecutor) threadPoolExecutor;
                Integer oldConfigId = etp.getActualConfig().getDynamicConfig()!=null?etp.getActualConfig().getDynamicConfig().getDynamicVersion()
                        :etp.getActualConfig().getFixedConfig().getFixedVersion();
                if(!etp.takeover() || etp.takeover() && takeover){
                    etp.updateConfig(limitConfig,suggestConfig);
                    if(takeover){
                        etp.confirmTakeover();
                    }
                }else{
                    return TpConfigUpdateRes.builder().extInfo("线程池被人工接管 自动切换不生效").success(false)
                            .build();
                }
                if(Boolean.parseBoolean(globalConfig.getHistory())){
                    if(persistentExecutor == null || persistentExecutor.isShutdown()){
                        persistentExecutor = createPersistentExecutor(globalConfig);
                    }
                    persistentExecutor.submit(()->{
                        this.persistentUpdateHistory(poolName,poolType,oldConfigId,suggestConfig.getDynamicConfig()!=null?
                                suggestConfig.getDynamicConfig().getDynamicVersion():suggestConfig.getFixedConfig().getFixedVersion(),userId);
                    });
                }else{
                    //否则需要判断一下 是否存在没有执行完的任务
                    if(persistentExecutor != null && !persistentExecutor.isShutdown()){
                        persistentExecutor.shutdown();
                    }
                }
            }else{
                threadPoolExecutor.setCorePoolSize(baseConfig.getPoolCoreSize());
                threadPoolExecutor.setMaximumPoolSize(baseConfig.getPoolMaxSize());
                threadPoolExecutor.setKeepAliveTime(baseConfig.getPoolKeepAliveTime(),
                        TimeUnit.valueOf(baseConfig.getPoolAliveTimeUnit().toString()));
                threadPoolExecutor.setRejectedExecutionHandler(baseConfig.getRejectHandle());
                if(threadPoolExecutor.getQueue() instanceof LinkedBlockingQueue){
                    ResizeableLinkedBlockingQueue.setCapacity((LinkedBlockingQueue<Runnable>)threadPoolExecutor.getQueue(),
                            baseConfig.getWorkQueueSize());
                }
            }

            return TpConfigUpdateRes.normal();
        } catch (Exception e) {
            e.printStackTrace();
            log.error("Fuys etp: 线程池更新配置失败 更新对象为:名称:{},类型:{},原因为:{}",poolName,poolType,e.getMessage());
            return TpConfigUpdateRes.builder().extInfo("更新失败").success(false).build();
        }
    }

    private void persistentUpdateHistory(String poolName,String poolType,Integer oldConfigId,
                                         Integer newConfigId,Long userId){
        log.info("Fuys Etp:记录配置变更:{},{},{},{},{}",poolName,poolType,oldConfigId,newConfigId,userId);
        etpRepoService.persistentUpdateHistory(poolName,poolType,oldConfigId,newConfigId,userId);
    }

    public TpStrategyChangeRes changeStrategy(TpStrategyChangeReq changeReq,boolean takeover){
        final String goalType = changeReq.getGoalType();
        if(EtpTypeEnum.COMMON_THREAD_POOL_EXECUTOR.type().equals(goalType)){
            return TpStrategyChangeRes.builder().extInfo("不支持此类型线程池")
                    .changeType(goalType).success(false).build();
        }
        final ThreadPoolExecutor threadPoolExecutor =
                executorMap.get(new Pair<>(goalType, changeReq.getGoalName()));
        if(threadPoolExecutor == null){
            return TpStrategyChangeRes.builder().extInfo("没有此线程池").success(false)
                    .changeType(goalType).build();
        }
        final EtpThreadPoolExecutor etp = (EtpThreadPoolExecutor) threadPoolExecutor;
        if(etp.takeover()){
            return TpStrategyChangeRes.builder().extInfo("处于接管状态 无法切换策略").success(false).build();
        }
        final String originType = etp.getType();
        final String name = etp.getName();
        TpRegulationStrategyVO strategy = strategyIdMap.get(changeReq.getStrategyId());
        if(strategy == null){
            strategy = etpRepoService.selectStrategyById(changeReq.getStrategyId().toString());
            if(strategy == null){
                return TpStrategyChangeRes.builder().extInfo("没有指定的策略").success(false).build();
            }
            sortedByLoadLevel(strategy.getStrategyType().toString()
                    ,strategy.getLoadConfigChain(),strategy);
            strategyMap.put(new Pair<>(changeReq.getGoalType(),changeReq.getGoalName()),strategy);
            strategyIdMap.put(strategy.getStrategyId(),strategy);
        }
        final Pair<TpConfigVO, TpConfigVO> pair = this.
                switchAndSaveConfig(strategy.getStrategyType().toString(), strategy);
        try {
            final TpStrategyChangeRes tpStrategyChangeRes = etp.changeStrategy(strategy.getStrategyType().toString()
                    , pair.getValue(),pair.getKey());
            if(!originType.equals(strategy.getStrategyType().toString())){
                executorMap.remove(new Pair<>(goalType,name));
                executorMap.put(new Pair<>(strategy.getStrategyType().toString(),name),etp);
            }
            return tpStrategyChangeRes;
        } catch (Exception e) {
            e.printStackTrace();
            return TpStrategyChangeRes.builder().success(false).extInfo(e.getMessage()).build();
        }
    }

    public ThreadPoolExecutor getExecutor(Pair<String, String> key) {
        return executorMap.get(key);
    }

    private ThreadPoolExecutor createPersistentExecutor(EtpPersistentConfig config){
        final int corePoolSize = Integer.parseInt(config.getPoolCoreSize());
        final int maximumPoolSize = Integer.parseInt(config.getPoolMaxSize());
        final long keepAliveTime = Long.parseLong(config.getPoolKeepAliveTime());
        final TimeUnit unit = TimeUnit.valueOf(config.getPoolAliveTimeUnit());
        return new ThreadPoolExecutor(corePoolSize,
                    maximumPoolSize, keepAliveTime,
                    unit, new LinkedBlockingQueue<>(),
                    new NamedThreadPoolFactory("Fuys-Etp-Persistent-Thread"));
    }

    private void updatePersistentExecutor(EtpPersistentConfig config){
        final int corePoolSize = Integer.parseInt(config.getPoolCoreSize());
        final int maximumPoolSize = Integer.parseInt(config.getPoolMaxSize());
        final long keepAliveTime = Long.parseLong(config.getPoolKeepAliveTime());
        final TimeUnit unit = TimeUnit.valueOf(config.getPoolAliveTimeUnit());
        if (this.persistentExecutor.getCorePoolSize() != corePoolSize) {
            this.persistentExecutor.setCorePoolSize(corePoolSize);
        }
        if(this.persistentExecutor.getMaximumPoolSize() != maximumPoolSize){
            this.persistentExecutor.setMaximumPoolSize(maximumPoolSize);
        }
        this.persistentExecutor.setKeepAliveTime(keepAliveTime,unit);
    }

    public void updatePersistentConfig(EtpPersistentConfig config) {
        if(Boolean.parseBoolean(config.getHistory())){
            //如果记录持久的线程池处于关闭状态 则开启
            if(this.persistentExecutor == null || this.persistentExecutor.isShutdown()){
                persistentExecutor = createPersistentExecutor(config);
            }else{
                this.updatePersistentExecutor(config);
            }
        }else {
            //则需要关闭持久化线程池
            this.persistentExecutor.shutdown();
        }
    }
}
